package com.ztrue.framework.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.ztrue.framework.infrastructure.IClone;
import com.ztrue.framework.infrastructure.IDistribute;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

/**
 * 表示一个消息队列
 *
 * @author zhangw
 * @date 2016/11/5 17:37
 * @version: 1.0
 */
public class MessageQueue<M extends IDistribute & IClone> implements IMessageQueue<M> {
    private final RingBuffer<M> ringBuffer;
    private final ExecutorService executors;

    MessageQueue(RingBuffer<M> ringBuffer, ExecutorService executors) {
        this.ringBuffer = ringBuffer;
        this.executors = executors;
    }

    /**
     * 发送一个消息
     *
     * @param consumer
     */
    public void send(Consumer<M> consumer) {
        executors.submit(new Callable<Void>() {
            public Void call() throws Exception {
                long seq = ringBuffer.next();
                M m = ringBuffer.get(seq);
                consumer.accept(m);
                ringBuffer.publish(seq);
                return null;
            }
        });
    }
}